/*
 *  ------------------------------------------------------------------
 *  Copyright © 2017 Hangzhou DtDream Technology Co.,Lt d. All rights reserved.
 *  ------------------------------------------------------------------
 *              Product: EMR
 *    Module Name: DataTrain
 *      Date Created: 2017-10-23
 *         Description:
 *  ------------------------------------------------------------------
 *  Modification History
 *  DATE            Name           Description
 *  ------------------------------------------------------------------
 *  2017-10-23     XZ 209
 *  ------------------------------------------------------------------
 */

package org.xukai.remoting.sdk.rpc.writer;

import com.dtdream.emr.transmission.job.Job;
import com.dtdream.emr.transmission.transfer.TransferNode;
import com.dtdream.emr.transmission.data.Record;
import com.dtdream.emr.transmission.exception.TransmissionException;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.serializers.CollectionSerializer;
import com.esotericsoftware.kryo.serializers.JavaSerializer;

import java.util.ArrayList;
import java.util.List;

import static com.dtdream.emr.transmission.exception.TransmissionException.Error.EXCEPTION_SDK_WRITER_WRITE;

public class BufferedWriter implements Writer {

    private Job job;
    private int id;
    private volatile boolean closed = false;

    private static int BUFF_SIZE = 10;
    private List<Record> recordList = new ArrayList<>();
    private TransferNode transferNode;

    private volatile TransmissionException emrException;

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public BufferedWriter(Job job, int id, TransferNode node) {
        this.job = job;
        this.id = id;
        this.transferNode = node;
    }

    private void writeAndClear() throws TransmissionException {
        if (null == emrException) {
            transferNode.writeAsync(job.getJobId(), id, recordList);
            recordList.clear();
        } else {
            throw emrException;
        }
    }

    public synchronized void write(Record record) throws TransmissionException {
        recordList.add(record);
        if (recordList.size() >= BUFF_SIZE) {
            writeAndClear();
        }
    }

    public synchronized void close() throws TransmissionException {
        closed = true;
        try {
            if (!recordList.isEmpty()) {
                writeAndClear();
            }
            transferNode.finishWriteAsync(job.getJobId(), id);

            if (null != emrException) {
                throw emrException;
            }
        } finally {
            job.transferClient.closeWriter(job.getJobId(), id, this.transferNode);
        }
    }

    public boolean isClosed() {
        return closed;
    }

    @Override
    public void setError(TransmissionException cause) {
        this.emrException = new TransmissionException(EXCEPTION_SDK_WRITER_WRITE, cause);
    }
}
